package org.example.mq.dataCenter;

import org.example.mq.common.Consumer;
import org.example.mq.common.ConsumerEnv;
import org.example.mq.common.MqException;
import org.example.mq.mqserver.VirtualHost;
import org.example.mq.mqserver.core.MSGQueue;
import org.example.mq.mqserver.core.Message;

import java.io.IOException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;


/*
* 实现用户消费消息的功能
* */
public class ConsumerManager {
    //将虚拟机和这个类绑定起来
    private VirtualHost parent;
    //构建一个线程池（专门用来执行消费者的消费功能）
    private ExecutorService workPool = Executors.newFixedThreadPool(4);
    //创建出一个队列，标示出那些队列增加了新的消息
    private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();
    //使用一个线程进行扫描(一直扫描队列，如果增加新的队列，将队列名和对应的方法提供给线程池)
    private Thread scannerThread = null;

    public ConsumerManager(VirtualHost parent) {
        this.parent = parent;
        //创建出扫描线程，不断的扫描队列，并消费其中的消息
        scannerThread = new Thread(()->{
           while (true){
               try{
                   String queueName = tokenQueue.take();
                   MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
                   if(queue==null){
                       throw new MqException("队列名不存在");
                   }
                   synchronized (queue){
                       //这里只会消费一个消息（来一个消息会添加一个队列名，来几个会添加几个队列名）
                       consumeMessage(queue);
                   }
               } catch (InterruptedException | MqException e) {
                   e.printStackTrace();
               }
           }
        });
        scannerThread.setDaemon(true);
        scannerThread.start();
    }

    public void notifyConsume(String queueName){
        tokenQueue.add(queueName);
    }

    /*
    * 订阅消息（实现的效果就是：如果这个队列增加消息，就告知消费者）
    * 增加绑定关系，并添加到queue队列中的consumerEnvs中
    * 获取消息的个数，并推送
    * */
    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
//        1.检查这个队列是否存在
        MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        if(queue == null){
            throw new MqException("队列不存在");
        }
//        2.创建出消费者和队列的绑定关系
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag,queueName,autoAck,consumer);
        synchronized (queue){
//            3.队列中添加这个消费者关系
            queue.addConsumerEnv(consumerEnv);
//            4.开始推送之前的和刚添加的队列中消息（干掉之前发送的消息和刚刚发送的消息）
            int n = parent.getMemoryDataCenter().getQueueMessageCount(queueName);
            for (int i = 0;i<n;i++){
                consumeMessage(queue);
            }
        }

    }

    /*
    * 推送给随机一个消费者推送队列中的消息（每次只消费一个）
    * */
    private void consumeMessage(MSGQueue queue) {
//        1.随机选择一个消费者出来（一个队列可以被多个消费者关注，这些消费者谁先开始有一个优先顺序，挑选一个）
        ConsumerEnv luckyDog = queue.chooseConsumer();
        if(luckyDog==null){
            return;
        }
//        2.开始推送消息
        //取出第一个消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if(message==null){
            return;
        }
//        3.这里推送消息的核心：将消息交给线程池，线程池去调用回调函数（具体的消费者想要执行的业务逻辑）
        //也就是去向线程池提交消息的回调函数，线程池会自动执行
        workPool.submit(()->{
            try{

//            1.先把消息放在待确认的集合中，不能直接执行回调函数，如果回调失败，则会出现消息丢失情况
                parent.getMemoryDataCenter().insertMessageWaitAck(queue,message);
//            2.执行回调函数
                luckyDog.getConsumer().handleDelivery(luckyDog.getConsumerTag(),message.getBasicProperties(),message.getBody());
//            3.如果是自动应答，则直接把消息进行删除即可
                //如果是手动应答，调用basicAck方法来处理
                if(luckyDog.isAutoAck()){
//                    应答表示已经接收到了消息，那么就需要删除硬盘和内存中的数据
                    if(message.getBasicProperties().getDeliverMode() == 2){
                        parent.getDiskDataCenter().deleteMessage(queue,message);
                    }
                    parent.getMemoryDataCenter().deleteMessageWaitAck(queue.getName(), message.getMessageId());
                    parent.getMemoryDataCenter().deleteMessage(message.getMessageId());
                }
                System.out.println("消息被成功消费");
            } catch (IOException | ClassNotFoundException | MqException e) {
                e.printStackTrace();
            }
        });
    }

}
